package com.shujia.spark.mllib

import org.apache.spark.ml.classification.LogisticRegressionModel
import org.apache.spark.ml.linalg
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo6imageModelUse {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName("source")
      .master("local[8]")
      .getOrCreate()

    //导入隐式转换
    import spark.implicits._
    //导入spark 所有的函数
    import org.apache.spark.sql.functions._


    //读取图片
    val imageData: DataFrame = spark
      .read
      .format("image")
      .load("D:\\课件\\机器学习数据\\手写数字\\test2")


    val featuresData: DataFrame = imageData
      .select($"image.origin" as "path", $"image.data" as "data")
      .map(row => {
        val path: String = row.getAs[String]("path")

        //取出文件名
        val name: String = path.split("/").last


        //取出图片的数据
        val data: Array[Byte] = row.getAs[Array[Byte]]("data")

        //处理数据，降噪
        val inData: Array[Double] = data.map(b => {
          val i: Int = b.toInt
          if (i < 0) {
            1.0
          } else {
            0.0
          }
        })

        val vector: linalg.Vector = Vectors.dense(inData)

        (name, vector)
      }).toDF("name", "features")

    /**
      * 加载模型
      *
      */

    val model: LogisticRegressionModel = LogisticRegressionModel.load("data/imageModel")


    //预测
    val frame: DataFrame = model.transform(featuresData)

    frame.show()

  }

}
